package com.hmeter.hadoop.hbase;

import java.io.IOException;

import org.apache.jmeter.samplers.AbstractSampler;
import org.apache.jmeter.samplers.Entry;
import org.apache.jmeter.samplers.SampleResult;

public class HBaseGetSampler extends AbstractSampler {
	
	/**
	 * 
	 */
	private static final long serialVersionUID = 6518225195340420663L;
	
	private static final String ZOOKEEPERQUORUM = "ZOOKEEPERQUORUM";
	private static final String HBASETABLE = "HBASETABLE";
	private static final String STARTROWKEY = "STARTROWKEY";
	private static final String STOPROWKEY = "STOPROWKEY";
	private static final String NUMBEROFROWSTORETRIEVE = "NUMBEROFROWSTORETRIEVE";
	private static final String FIRSTCHECKBOX = "FIRSTCHECKBOX";
	private static final String FIRSTCOLUMNFAMILYTEXTBOX = "FIRSTCOLUMNFAMILYTEXTBOX";
	private static final String FIRSTCOLUMNQUALIFIERTEXTBOX = "FIRSTCOLUMNQUALIFIERTEXTBOX";
	private static final String SECONDCHECKBOX = "SECONDCHECKBOX";
	private static final String SECONDCOLUMNFAMILYTEXTBOX = "SECONDCOLUMNFAMILYTEXTBOX";
	private static final String THIRDCHECKBOX = "THIRDCHECKBOX";
	private static final String FOURTHCHECKBOX = "FOURTHCHECKBOX";
	private static final String FIFTHCHECKBOX = "FIFTHCHECKBOX";
	private static final String HBASEFILTER = "HBASEFILTER";
	private static final String HBASEOPERATOR = "HBASEOPERATOR";
	private static final String DELIMITER = "DELIMITER";
	private static final String STOPROWKEYBOOL = "STOPROWKEYBOOL";
	private static final String NUMBEROFROWSTORETRIEVEBOOL = "NUMBEROFROWSTORETRIEVEBOOL";
	
	@Override
	public SampleResult sample(Entry e) {
		SampleResult result = new SampleResult();
		
		String requestData = "zookeeper quorum: " + this.getZookeeperQuorum() + "\n";
		requestData += "hbase table: " + this.getHBaseTable() + "\n";
		requestData += "start row key: " + this.getStartRowKey() + "\n";
		requestData += "end row key: " + this.getStopRowKey() + "\n";
		requestData += "number of rows to retrieve: " + this.getNumberOfRowsToRetrieve() + "\n";
		requestData += "hbase filters: " + this.getHBaseFilter() + "\n";
		requestData += "hbase operator: " + this.getHBaseOperator() + "\n";
		requestData += "hbase filters delimiter: " + this.getDelimiter() + "\n";
		requestData += "column family to retrieve: " + this.getSecondColumnFamily() + "\n";
		requestData += "column family and column qualifier to retrieve: " + this.getFirstColumnFamily() + " , " + this.getFirstColumnQualifier() + "\n";
		
		
		result.setSampleLabel(getName());
		result.setSamplerData(requestData);
		result.setDataType(SampleResult.TEXT);

		result.sampleStart();
		
		int numberOfRows = 0;
		HBaseGetSamplerUtility hbaseGet = null;
		
		try {
			hbaseGet = new HBaseGetSamplerUtility(this.getZookeeperQuorum(), this.getHBaseTable());

			if (!this.getNumberOfRowsToRetrieve().equals(""))
			{
				numberOfRows = Integer.parseInt(this.getNumberOfRowsToRetrieve());
			}

		} catch (IOException e1) {
			result.setResponseData(e1.getMessage().getBytes());
			result.setSuccessful(false);
			System.out.println(e1.getMessage());
		} catch (NumberFormatException e2) {
			result.setResponseData(e2.getMessage().getBytes());
			result.setSuccessful(false);
			System.out.println(e2.getMessage());
		}
		
		if (this.isFirstCheckbox())
		{
			if (this.isStopRowKeyBool())
			{
				try {
					String response = hbaseGet.getColumnQualifierForRangeOfRows(this.getStartRowKey(), this.getStopRowKey(), this.getFirstColumnFamily(), this.getFirstColumnQualifier());
					result.setResponseData(response.getBytes());
					result.setSuccessful(true);
				} catch (IOException e1) {
					result.setResponseData(e1.getMessage().getBytes());
					result.setSuccessful(false);
					System.out.println(e1.getMessage());
				}
			}
			
			if (this.isNumberOfRowsToRetrieveBool())
			{
				try {
					String response = hbaseGet.getColumnQualifierForRangeOfRows_NumberOfRowsToRetrieve(this.getStartRowKey(), numberOfRows, this.getFirstColumnFamily(), this.getFirstColumnQualifier());
					result.setResponseData(response.getBytes());
					result.setSuccessful(true);
					System.out.println(numberOfRows);
				} catch (IOException e1) {
					result.setResponseData(e1.getMessage().getBytes());
					result.setSuccessful(false);
					System.out.println(e1.getMessage());
				}
			}
		}
		
		if (this.isSecondCheckbox())
		{
			if (this.isStopRowKeyBool())
			{
				try {
					String response = hbaseGet.getColumnQualifiersForRangeOfRowsByColumnFamily(this.getStartRowKey(), this.getStopRowKey(), this.getSecondColumnFamily());
					result.setResponseData(response.getBytes());
					result.setSuccessful(true);
				} catch (IOException e1) {
					result.setResponseData(e1.getMessage().getBytes());
					result.setSuccessful(false);
				}
			}
			
			if (this.isNumberOfRowsToRetrieveBool())
			{
				try {
					String response = hbaseGet.getColumnQualifiersForRangeOfRowsByColumnFamily_NumberOfRowsToRetrieve(this.getStartRowKey(), numberOfRows, this.getSecondColumnFamily());
					result.setResponseData(response.getBytes());
					result.setSuccessful(true);
				} catch (IOException e1) {
					result.setResponseData(e1.getMessage().getBytes());
					result.setSuccessful(false);
				}
			}
		}
		
		if (this.isThirdCheckbox())
		{
			if (this.isStopRowKeyBool())
			{
				try {
					String response = hbaseGet.getAllColumnQualifiersForRangeOfRows(this.getStartRowKey(), this.getStopRowKey());
					result.setResponseData(response.getBytes());
					result.setSuccessful(true);
				} catch (IOException e1) {
					result.setResponseData(e1.getMessage().getBytes());
					result.setSuccessful(false);
				}
			}
			
			if (this.isNumberOfRowsToRetrieveBool())
			{
				try {
					String response = hbaseGet.getAllColumnQualifiersForRangeOfRows_NumberOfRowsToRetrieve(this.getStartRowKey(), numberOfRows);
					result.setResponseData(response.getBytes());
					result.setSuccessful(true);
				} catch (IOException e1) {
					result.setResponseData(e1.getMessage().getBytes());
					result.setSuccessful(false);
				}
			}
		}
		
		if (this.isFourthCheckbox())
		{
			if (this.isStopRowKeyBool())
			{
				try {
					String response = hbaseGet.getAllColumnQualifiersForRangeOfRowsByMostRecentTimestamp(this.getStartRowKey(), this.getStopRowKey());
					result.setResponseData(response.getBytes());
					result.setSuccessful(true);
				} catch (IOException e1) {
					result.setResponseData(e1.getMessage().getBytes());
					result.setSuccessful(false);
				}
			}
			
			if (this.isNumberOfRowsToRetrieveBool())
			{
				try {
					String response = hbaseGet.getAllColumnQualifiersForRangeOfRowsByMostRecentTimestamp_NumberOfRowsToRetrieve(this.getStartRowKey(), numberOfRows);
					result.setResponseData(response.getBytes());
					result.setSuccessful(true);
				} catch (IOException e1) {
					result.setResponseData(e1.getMessage().getBytes());
					result.setSuccessful(false);
				}
			}
		}
		
		if (this.isFifthCheckbox())
		{
			if (this.isStopRowKeyBool())
			{
				try {
					String response = hbaseGet.getRowsContentByFilters(this.getStartRowKey(), this.getStopRowKey(), this.getHBaseOperator(), this.getHBaseFilter(), this.getDelimiter());
					result.setResponseData(response.getBytes());
					result.setSuccessful(true);
				} catch (Exception e1) {
					result.setResponseData(e1.getMessage().getBytes());
					result.setSuccessful(false);
				}
			}
			
			if (this.isNumberOfRowsToRetrieveBool())
			{
				try {
					String response = hbaseGet.getRowsContentByFilters_NumberOfRowsToRetrieve(this.getStartRowKey(), numberOfRows, this.getHBaseOperator(), this.getHBaseFilter(), this.getDelimiter());
					result.setResponseData(response.getBytes());
					result.setSuccessful(true);
				} catch (Exception e1) {
					result.setResponseData(e1.getMessage().getBytes());
					result.setSuccessful(false);
				}
			}
		}
		
		result.sampleEnd();
		return result;
	}
	
	public void setZookeeperQuorum(String zookeeperQuorum)
	{
		setProperty(ZOOKEEPERQUORUM, zookeeperQuorum);
	}
	
	public String getZookeeperQuorum()
	{
		return getPropertyAsString(ZOOKEEPERQUORUM);
	}
	
	public void setHBaseTable(String hbaseTable)
	{
		setProperty(HBASETABLE, hbaseTable);
	}
	
	public String getHBaseTable()
	{
		return getPropertyAsString(HBASETABLE);
	}
	
	public void setStartRowKey(String startRowKey)
	{
		setProperty(STARTROWKEY, startRowKey);
	}
	
	public String getStartRowKey()
	{
		return getPropertyAsString(STARTROWKEY);
	}
	
	public void setStopRowKey(String stopRowKey)
	{
		setProperty(STOPROWKEY, stopRowKey);
	}
	
	public String getStopRowKey()
	{
		return getPropertyAsString(STOPROWKEY);
	}
	
	public void setNumberOfRowsToRetrieve(String numOfRows)
	{
		setProperty(NUMBEROFROWSTORETRIEVE, numOfRows);
	}
	
	public String getNumberOfRowsToRetrieve()
	{
		return getPropertyAsString(NUMBEROFROWSTORETRIEVE);
	}
		
	public void setFirstColumnFamily(String firstColumnFamily)
	{
		setProperty(FIRSTCOLUMNFAMILYTEXTBOX, firstColumnFamily);
	}
	
	public String getFirstColumnFamily()
	{
		return getPropertyAsString(FIRSTCOLUMNFAMILYTEXTBOX);
	}

	public void setFirstColumnQualifier(String firstColumnQualifier)
	{
		setProperty(FIRSTCOLUMNQUALIFIERTEXTBOX, firstColumnQualifier);
	}
	
	public String getFirstColumnQualifier()
	{
		return getPropertyAsString(FIRSTCOLUMNQUALIFIERTEXTBOX);
	}
	
	public void setSecondColumnFamily(String secondColumnFamily)
	{
		setProperty(SECONDCOLUMNFAMILYTEXTBOX, secondColumnFamily);
	}
	
	public String getSecondColumnFamily()
	{
		return getPropertyAsString(SECONDCOLUMNFAMILYTEXTBOX);
	}
	
	public void setHBaseFilter(String hbaseFilter)
	{
		setProperty(HBASEFILTER, hbaseFilter);
	}
	
	public String getHBaseFilter()
	{
		return getPropertyAsString(HBASEFILTER);
	}
	
	public void setHBaseOperator(String hbaseOperator)
	{
		setProperty(HBASEOPERATOR, hbaseOperator);
	}
	
	public String getHBaseOperator()
	{
		return getPropertyAsString(HBASEOPERATOR);
	}
		
	public void setDelimiter(String delimiter)
	{
		setProperty(DELIMITER, delimiter);
	}
	
	public String getDelimiter()
	{
		return getPropertyAsString(DELIMITER);
	}
	
	public boolean isFirstCheckbox()
	{
		return getPropertyAsBoolean(FIRSTCHECKBOX);
	}
	
	public void setFirstCheckbox(boolean firstCheckBox)
	{
		setProperty(FIRSTCHECKBOX, firstCheckBox);
	}
	
	public boolean isSecondCheckbox()
	{
		return getPropertyAsBoolean(SECONDCHECKBOX);
	}
	
	public void setSecondCheckbox(boolean secondCheckBox)
	{
		setProperty(SECONDCHECKBOX, secondCheckBox);
	}
	
	public boolean isThirdCheckbox()
	{
		return getPropertyAsBoolean(THIRDCHECKBOX);
	}
	
	public void setThirdCheckbox(boolean thirdCheckBox)
	{
		setProperty(THIRDCHECKBOX, thirdCheckBox);
	}
	
	public boolean isFourthCheckbox()
	{
		return getPropertyAsBoolean(FOURTHCHECKBOX);
	}
	
	public void setFourthCheckbox(boolean fourthCheckBox)
	{
		setProperty(FOURTHCHECKBOX, fourthCheckBox);
	}
	
	public boolean isFifthCheckbox()
	{
		return getPropertyAsBoolean(FIFTHCHECKBOX);
	}
	
	public void setFifthCheckbox(boolean fifthCheckBox)
	{
		setProperty(FIFTHCHECKBOX, fifthCheckBox);
	}
	
	public boolean isStopRowKeyBool()
	{
		return getPropertyAsBoolean(STOPROWKEYBOOL);
	}
	
	public void setStopRowKeyBool(boolean stopRowKeyBool)
	{
		setProperty(STOPROWKEYBOOL, stopRowKeyBool);
	}
	
	public boolean isNumberOfRowsToRetrieveBool()
	{
		return getPropertyAsBoolean(NUMBEROFROWSTORETRIEVEBOOL);
	}
	
	public void setNumberOfRowsToRetrieveBool(boolean numberOfRowsToRetrieveBool)
	{
		setProperty(NUMBEROFROWSTORETRIEVEBOOL, numberOfRowsToRetrieveBool);
	}
}
